package com.hzgj.bcl.soa.ws;

import com.google.common.collect.Maps;
import com.hzgj.bcl.util.lang.StrKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public abstract class MessageHandler extends TextWebSocketHandler {

    protected Logger logger = LoggerFactory.getLogger(getClass());
    Map<WebSocketSession, String> session2Key = Maps.newHashMap();
    BlockingQueue<byte[]> queue;
    Thread loop;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        logger.debug("connect to the websocket success......");
        queue = new LinkedBlockingQueue<>();
        loop = loop_send(queue, session);
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        super.handleTextMessage(session, message);
        String parameters = message.getPayload();
        if (StrKit.notBlank(parameters)) {
            session2Key.put(session, parameters);
            handle(parameters, queue);
        }
    }

    abstract void handle(String parameters, BlockingQueue<byte[]> queue);

    abstract String getName();

    Thread loop_send(BlockingQueue<byte[]> queue, WebSocketSession session) {
        Thread thread = new Thread(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    byte[] result = queue.take();
                    sendMessage(new TextMessage(result), session);
                }
            } catch (InterruptedException e1) {
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                logger.error("{}", e);
            }
        });
        thread.setDaemon(true);
        thread.start();
        thread.setName("loop-send-" + getName());
        return thread;
    }

    public void sendMessage(TextMessage message, WebSocketSession session) {
        try {
            if (session.isOpen()) {
                session.sendMessage(message);
            }
        } catch (IOException e) {
            logger.error("{}", e);
        }
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            session.close();
        }
        logger.debug("websocket connection closed......");
        destroy(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        logger.debug("websocket connection closed......");
        destroy(session);
    }

    void destroy(WebSocketSession session) {
        String key = session2Key.remove(session);
        if (StrKit.notBlank(key)) {
            loop.interrupt();
        }
        try {
            session.close();
        } catch (IOException e) {
            logger.error("{}", e);
        }
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }
}